#include <iostream>
#include <stdio.h>
#include <string>
#include <unistd.h>
#include <dirent.h>
#include <string.h>
#include <sys/time.h>
#include <pthread.h>
#include <vector>
#include <string>
#include "../buttonrpc/buttonrpc.hpp"
#include <fstream> // 包含文件流相关操作
#include <sstream> // 包含字符串流相关操作
#include <cstdlib>
#include <unordered_map>
#include <algorithm>


using namespace std;

struct Task
{
    string fileName ;
    int IdMap;
// -------------
    int IdReduce;
};

// 回馈消息
struct TaskResponse{
    int  state;             // 当前标识
    Task curTask;           // 当前任务
    int  curNumMapTask;     // 当前待分配map任务数量
    int  curNumReduceTask;  // 当前待分配reduce任务数量
    int  NumMapTask;        // 总map任务数量
    int  NumReduceTask;     // 总reduce任务数量
}; 


struct KeyValue{
    string key;
    string value;
};

/**
 * @brief 字符串分割函数
 * 
 * @param text 传入的文本内容，数据量极大
 * @param length 传入的字符串长度
 * @return vector<string> 返回各个分割后的单词列表
 */
vector<string> split(const char* text, int length){
    vector<string> str;
    string tmp = "";
    fflush(stdout);                     // 清空缓冲区，否则多次运行会出现bug
    for(int i = 0; i < length; i++){
        if((text[i] >= 'A' && text[i] <= 'Z') || (text[i] >= 'a' && text[i] <= 'z')){
            tmp += text[i];
        }else{
            if(tmp.size() != 0) str.emplace_back(tmp);      
            tmp = "";
        }
    }
    if(tmp.size() != 0) str.emplace_back(tmp);
    return str;
}

//以char类型的op为分割拆分字符串
vector<string> split(string text, char op){
    int n = text.size();
    vector<string> str;
    string tmp = "";
    for(int i = 0; i < n; i++){
        if(text[i] != op){
            tmp += text[i];
        }else{
            if(tmp.size() != 0) str.push_back(tmp);
            tmp = "";
        }
    }
    return str;
}

//以逗号为分割拆分字符串
string split(string text){
    string tmp = "";
    for(int i = 0; i < text.size(); i++){
        if(text[i] != ','){
            tmp += text[i];
        }else break;
    }
    return tmp;
}

/**
 * @brief kv 将文本按单词划分并以出现次数代表value长度存入keyValue
 * @return 类似{"my 11111", "you 111"} 即文章中my出现了5次，you出现了3次
 */
vector<KeyValue> mapF(KeyValue& kv){
    vector<KeyValue> kvs;
    int len = kv.value.size();
    char content[len + 1];
    strcpy(content, kv.value.c_str());
    vector<string> str = split(content, len);
    for(const auto& s : str){
        KeyValue tmp;
        tmp.key = s;
        tmp.value = "1";
        kvs.emplace_back(tmp);
    }
    return kvs;
}

/**
 * @brief 返回单词的数量
 * @return vector<string>
 */
vector<string> reduceF(vector<KeyValue> kvs){
    vector<string> str;
    string tmp;
    for(const auto& kv : kvs){
        str.emplace_back(to_string(kv.value.size()));
    }
    return str;
}

//对每个字符串求hash找到其对应要分配的reduce线程
int ihash(string key){
    int sum = 0;
    for(int i = 0; i < key.size(); i++){
        sum += (key[i] - '0');
    }
    return sum;
}

//删除所有写入中间值的临时文件
void removeFiles(int map_task_num, int reduce_task_num){
    string path;
    for(int i = 0; i < map_task_num; i++){
        for(int j = 0; j < reduce_task_num; j++){
            path = "mr-" + to_string(i) + "-" + to_string(j);
            int ret = access(path.c_str(), F_OK);
            if(ret == 0) remove(path.c_str());
            else cerr << "access failed!";
        }
    }
}

// 获取文件内容
KeyValue getContent(const std::string& file) {
    std::ifstream inputFile(file);
    if (!inputFile.is_open()) {
        std::cerr << "Failed to open file: " << file << std::endl;
        exit(-1);
    }

    std::stringstream buffer;
    buffer << inputFile.rdbuf();
    std::string fileContent = buffer.str();

    KeyValue kv;
    kv.key = file;
    kv.value = fileContent;

    inputFile.close();
    return kv;
}

void WirteOutReduce(){
    
}

// 中间文件 map
void writeOutMap(const std::vector<KeyValue>& kvs, int mapTaskId, int NumReduceTask) {
    for (const auto& v : kvs) {
        int reduce_idx = ihash(v.key) % NumReduceTask;
        std::string path = "mr-" + std::to_string(mapTaskId) + "-" + std::to_string(reduce_idx);

        std::ofstream outfile(path, std::ios::out | std::ios::app);
        if (!outfile.is_open()) {
            std::cerr << "Failed to open file: " << path << std::endl;
            exit(-1);
        } else {
            std::string tmp = v.key + ",1 ";
            outfile << tmp;
        }
        outfile.close();
    }
}

//获取对应reduce编号的所有中间文件
vector<string> getAllfile(string path, int op){
    DIR *dir = opendir(path.c_str());
    vector<string> ret;
    if (dir == NULL)
    {
        printf("[ERROR] %s is not a directory or not exist!", path.c_str());
        return ret;
    }
    struct dirent* entry;
    while ((entry=readdir(dir)) != NULL)
    {
        int len = strlen(entry->d_name);
        int oplen = to_string(op).size();
        if(len - oplen < 5) continue;
        string filename(entry->d_name);
        if(!(filename[0] == 'm' && filename[1] == 'r' && filename[len - oplen - 1] == '-')) continue;
        string cmp_str = filename.substr(len - oplen, oplen);
        if(cmp_str == to_string(op)){
            ret.push_back(entry->d_name);
        }
    }
    closedir(dir);
    return ret;
}

// 对于一个ReduceTask，获取所有相关文件并将Keyvalue的进行排序
// vector中每个元素的形式为"abc 11111";
vector<KeyValue> shuffle(int reduceTaskNum){
    string path;
    vector<string> str;
    str.clear();
    vector<string> filename = getAllfile(".", reduceTaskNum);
    unordered_map<string, string> hash;
    for(int i = 0; i < filename.size(); i++){
        path = filename[i];
        KeyValue kv = getContent(path);
        string context = kv.value;
        vector<string> retStr = split(context, ' ');
        str.insert(str.end(), retStr.begin(), retStr.end());
    }
    for(const auto& s : str){
        hash[split(s)] += "1";
    }
    vector<KeyValue> retKvs;
    KeyValue tmpKv;
    for(const auto& a : hash){
        tmpKv.key = a.first;
        tmpKv.value = a.second;
        retKvs.push_back(tmpKv);
    }
    sort(retKvs.begin(), retKvs.end(), [](KeyValue& kv1, KeyValue& kv2){
        return kv1.key < kv2.key;
    });
    return retKvs;
}

void Worker(){
    buttonrpc client;
    client.as_client("127.0.0.1", 5555);

    while (1){
        TaskResponse reply;
        reply = client.call<TaskResponse>("AssignTask").val();

        int state = reply.state;
        if (reply.curNumMapTask >= 0 && state == 0){   // 继续完成map任务
            string filename = reply.curTask.fileName;
            KeyValue kv = getContent(filename);

            // 执行map函数
            vector<KeyValue> kvs = mapF(kv);
            writeOutMap(kvs, reply.curTask.IdMap, reply.NumReduceTask);

            // 发送RPC给Coordinator告知已完成map任务的id
            client.call<void*>("TaskFin", reply.curTask.IdMap);

        } else if (reply.curNumReduceTask >= 0 && state == 1) {   // 继续完成reduce任务
            //取得reduce任务，读取对应文件，shuffle后调用reduceFunc进行reduce处理
            vector<KeyValue> kvs = shuffle(reply.curTask.IdReduce);
            vector<string> ret = reduceF(kvs);
            vector<string> str;
            for(int i = 0; i < kvs.size(); i++){
                str.emplace_back(kvs[i].key + " " + ret[i]);
            }
            string path = "mr-out-" + to_string(reply.curTask.IdReduce);
            std::ofstream outfile(path, std::ios::out | std::ios::app);
            if (!outfile.is_open()) {
                std::cerr << "Failed to open file: " << path << std::endl;
                exit(-1);
            } else {
                for(auto s : str){
                    outfile << s << endl;
                }
            }
            outfile.close();

            // 发送RPC给Coordinator告知已完成reduce任务的id
            client.call<void*>("TaskFin", reply.curTask.IdReduce);
        } else if (state == 2) {
            break;
        } else {
            continue;
        }
        sleep(1);
    }
    
}

int main(){
    Worker();
}